package com.xieshue.im.mq;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;

/**
 * Created by Administrator on 2018/1/14.
 */
public class ConnectionMq {

    private static final Logger logger = LoggerFactory.getLogger(ConnectionMq.class);
    private static ConnectionMq ourInstance = new ConnectionMq();

    public static ConnectionMq getInstance() {
        return ourInstance;
    }

    private ConnectionMq() {
        init();
    }

    private void init() {
        this.factory = createFactory();
        this.setAdmin(factory);
    }

    private void setAdmin(ConnectionFactory factory) {
        this.admin = new RabbitAdmin(factory);
        declareQueue();
        declareExchange();
        declareBinding();
    }
    private void declareQueue() {
        this.queue = new Queue("myQueue");
        admin.declareQueue(queue);
    }

    private void declareExchange() {
        exchange = new TopicExchange("myExchange");
        admin.declareExchange(exchange);
    }

    private void declareBinding() {
        // set up the queue, exchange, binding on the broker
        admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo.*"));
    }


    private ConnectionFactory factory;
    private RabbitAdmin admin;

    private Queue queue;
    private TopicExchange exchange;
    private SimpleMessageListenerContainer container;

    public ConnectionFactory createFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setAddresses("127.0.0.1:5672");
        factory.setUsername("spxse");
        factory.setPassword("1234");
        factory.setVirtualHost("/");
        factory.setPublisherConfirms(true); //必须要设置
        return factory;
    }

    public void registerConsumer(SimpleMessageListenerContainer container) {
        // set up the listener and container
        this.container = container;
        this.container.setConnectionFactory(factory);
        this.container.setQueues(queue);
        this.container.setExposeListenerChannel(true);
        this.container.setMaxConcurrentConsumers(1);
        this.container.setConcurrentConsumers(1);
        this.container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        Object listener = new Object(){
            public void handleMessage(String foo) {
                System.out.println(false);
            }
        };
        MessageListenerAdapter adapter = new MessageListenerAdapter(listener);
//        container.setMessageListener(adapter);
        this.container.setMessageListener(new ChannelAwareMessageListener(){
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                logger.info("消费端接收到消息 : " + new String(body));
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        });
        this.container.setQueueNames("myQueue");
        this.container.start();
    }

    public SimpleMessageListenerContainer getContainer() {
        return container;
    }
}
